#例【2-1】
from typing import Optional, Dict

class ContextObject:
    def __init__(
        self,
        role: str,
        content: str,
        name: Optional[str] = None,
        status: Optional[str] = None,
        metadata: Optional[Dict] = None,
        id: Optional[str] = None,
        parent_id: Optional[str] = None
    ):
        self.role = role  # 角色类型，如 user, assistant, system, tool
        self.name = name  # 函数名或工具名（可选）
        self.content = content  # 实际Prompt内容
        self.status = status  # 状态标签：locked, sampled, readonly, hidden
        self.metadata = metadata or {}  # 扩展元信息
        self.id = id  # 唯一标识（可选）
        self.parent_id = parent_id  # 父级上下文引用（可选）

    def to_dict(self):
        return {
            "role": self.role,
            "name": self.name,
            "content": self.content,
            "status": self.status,
            "metadata": self.metadata,
            "id": self.id,
            "parent_id": self.parent_id
        }



#例【2-2】
from typing import Optional, Dict

class PromptUnit:
    def __init__(
        self,
        role: str,
        content: str,
        status: Optional[str] = None,
        metadata: Optional[Dict] = None,
        id: Optional[str] = None
    ):
        self.role = role                    # user / assistant / system / tool
        self.content = content              # 实际语义内容
        self.status = status or "active"    # 状态如 locked / hidden / readonly
        self.metadata = metadata or {}      # 附加边界控制信息，如token_count
        self.id = id                        # 唯一标识，用于边界控制定位

    def is_visible(self) -> bool:
        return self.status not in ["hidden", "deleted"]

    def is_editable(self) -> bool:
        return self.status not in ["locked", "readonly"]

    def to_dict(self):
        return {
            "role": self.role,
            "content": self.content,
            "status": self.status,
            "metadata": self.metadata,
            "id": self.id
        }



#例【2-3】
from typing import Optional, List, Dict

class ContextObject:
    def __init__(
        self,
        id: str,
        content: str,
        role: str,
        parent_id: Optional[str] = None,
        status: Optional[str] = None,
        metadata: Optional[Dict] = None
    ):
        self.id = id
        self.content = content
        self.role = role
        self.parent_id = parent_id  # 上一个上下文节点
        self.status = status or "active"
        self.metadata = metadata or {}

    def to_dict(self):
        return {
            "id": self.id,
            "content": self.content,
            "role": self.role,
            "parent_id": self.parent_id,
            "status": self.status,
            "metadata": self.metadata
        }

class ContextChain:
    def __init__(self):
        self.nodes: Dict[str, ContextObject] = {}

    def add_prompt(self, prompt: ContextObject):
        self.nodes[prompt.id] = prompt

    def trace_chain(self, start_id: str) -> List[ContextObject]:
        chain = []
        current = self.nodes.get(start_id)
        while current:
            chain.insert(0, current)
            current = self.nodes.get(current.parent_id)
        return chain



#例【2-4】
from typing import List, Dict, Optional

class ContextObject:
    def __init__(self, role: str, content: str, status: Optional[str] = None):
        self.role = role
        self.content = content
        self.status = status or "active"

    def to_dict(self):
        return {
            "role": self.role,
            "content": self.content,
            "status": self.status
        }

class SharedContext:
    def __init__(self, context_id: str, prompts: List[ContextObject]):
        self.context_id = context_id
        self.prompts = prompts  # 一段可被跨模型复用的上下文结构

    def filter_for_model(self, model_capability: str) -> List[ContextObject]:
        # 仅保留该模型能力范围内的Prompt
        return [p for p in self.prompts if model_capability in p.status]

    def to_model_input(self) -> List[Dict]:
        return [p.to_dict() for p in self.prompts]



#例【2-5】
import json
from typing import Dict, List
import copy
import uuid

class ContextObject:
    def __init__(self, role, content, parent_id=None, status="active"):
        self.id = str(uuid.uuid4())
        self.role = role
        self.content = content
        self.parent_id = parent_id
        self.status = status

    def to_dict(self) -> Dict:
        return {
            "id": self.id,
            "role": self.role,
            "content": self.content,
            "parent_id": self.parent_id,
            "status": self.status
        }

    @classmethod
    def from_dict(cls, d):
        obj = cls(d["role"], d["content"], d.get("parent_id"), d.get("status", "active"))
        obj.id = d["id"]
        return obj

class ContextChain:
    def __init__(self):
        self.prompts: Dict[str, ContextObject] = {}

    def add(self, prompt: ContextObject):
        self.prompts[prompt.id] = prompt

    def trace(self, end_id: str) -> List[ContextObject]:
        result = []
        current = self.prompts.get(end_id)
        while current:
            result.insert(0, current)
            current = self.prompts.get(current.parent_id)
        return result

    def create_snapshot(self, end_id: str) -> str:
        chain = self.trace(end_id)
        data = [p.to_dict() for p in chain]
        return json.dumps(data, indent=2, ensure_ascii=False)

    def restore_snapshot(self, snapshot_json: str):
        self.prompts.clear()
        data = json.loads(snapshot_json)
        for item in data:
            obj = ContextObject.from_dict(item)
            self.add(obj)
        return data[-1]["id"] if data else None



#例【2-6】
import time
from typing import Dict, Optional
import uuid

class ContextObject:
    def __init__(self, role, content, parent_id=None, status="active", metadata=None):
        self.id = str(uuid.uuid4())
        self.role = role
        self.content = content
        self.parent_id = parent_id
        self.status = status
        self.metadata = metadata or {}

    def to_dict(self):
        return {
            "id": self.id,
            "role": self.role,
            "content": self.content,
            "parent_id": self.parent_id,
            "status": self.status,
            "metadata": self.metadata
        }

class ExecutionEngine:
    def __init__(self):
        self.contexts: Dict[str, ContextObject] = {}

    def add_context(self, ctx: ContextObject):
        self.contexts[ctx.id] = ctx

    def execute(self, ctx_id: str):
        ctx = self.contexts.get(ctx_id)
        if not ctx:
            print(f"[ERROR] 无效ID: {ctx_id}")
            return

        if ctx.status == "interrupted":
            print(f"[WAIT] Prompt '{ctx.content[:20]}...' 被中断，等待恢复")
        elif ctx.status == "deferred":
            delay = ctx.metadata.get("defer_seconds", 5)
            print(f"[DEFER] Prompt延迟 {delay} 秒执行: {ctx.content[:20]}...")
            time.sleep(delay)
            ctx.status = "active"
            print(f"[RUN] 继续执行: {ctx.content[:20]}...")
        elif ctx.status == "active":
            print(f"[EXEC] 执行: {ctx.role} 说：{ctx.content}")
        else:
            print(f"[SKIP] 状态为 {ctx.status}，跳过")

    def resume(self, ctx_id: str):
        ctx = self.contexts.get(ctx_id)
        if ctx and ctx.status == "interrupted":
            ctx.status = "active"
            print(f"[RESUME] 已恢复中断 Prompt: {ctx.content[:20]}")

# 示例流程
if __name__ == "__main__":
    engine = ExecutionEngine()

    # 正常Prompt
    p1 = ContextObject(role="user", content="请帮我生成日报")
    engine.add_context(p1)

    # 中断Prompt
    p2 = ContextObject(role="assistant", content="正在调用外部数据源", parent_id=p1.id, status="interrupted")
    engine.add_context(p2)

    # 延迟Prompt
    p3 = ContextObject(role="assistant", content="已获取数据，准备生成摘要", parent_id=p2.id, status="deferred", metadata={"defer_seconds": 3})
    engine.add_context(p3)

    # 执行流程
    print("=== 初始执行阶段 ===")
    engine.execute(p1.id)
    engine.execute(p2.id)
    engine.execute(p3.id)

    # 模拟恢复
    print("\n=== 外部依赖完成，恢复执行 ===")
    engine.resume(p2.id)
    engine.execute(p2.id)
    engine.execute(p3.id)



#例【2-7】
from typing import Callable, Dict, List
import uuid
import time

# 定义Prompt对象
class Prompt:
    def __init__(self, content, role="user", status="active"):
        self.id = str(uuid.uuid4())
        self.content = content
        self.role = role
        self.status = status

# 状态变更事件
class StateChangeEvent:
    def __init__(self, prompt_id, old_state, new_state):
        self.prompt_id = prompt_id
        self.old_state = old_state
        self.new_state = new_state
        self.timestamp = time.time()

# 发布-订阅系统
class StateChangeNotifier:
    def __init__(self):
        self.subscribers: Dict[str, List[Callable[[StateChangeEvent], None]]] = {}

    def subscribe(self, event_type: str, handler: Callable[[StateChangeEvent], None]):
        self.subscribers.setdefault(event_type, []).append(handler)

    def notify(self, event_type: str, event: StateChangeEvent):
        for handler in self.subscribers.get(event_type, []):
            handler(event)

# 初始化系统
notifier = StateChangeNotifier()

# 示例监听器1：打印提示
def print_handler(event: StateChangeEvent):
    print(f"[事件通知] Prompt状态从 {event.old_state} → {event.new_state}，时间戳：{event.timestamp:.2f}")

# 示例监听器2：模拟数据库更新
def database_handler(event: StateChangeEvent):
    print(f"[DB更新] Prompt {event.prompt_id[:6]} 已进入状态：{event.new_state}，同步至数据库")

# 订阅事件
notifier.subscribe("state_changed", print_handler)
notifier.subscribe("state_changed", database_handler)

# 执行Prompt状态变更并发送通知
def update_prompt_status(prompt: Prompt, new_status: str):
    old_status = prompt.status
    prompt.status = new_status
    event = StateChangeEvent(prompt.id, old_status, new_status)
    notifier.notify("state_changed", event)

# 示例执行
if __name__ == "__main__":
    p = Prompt(content="请帮我生成工作周报")
    print(f"[初始状态] Prompt: {p.content}，状态：{p.status}")
    
    update_prompt_status(p, "sampled")   # 模拟模型已生成初稿
    update_prompt_status(p, "locked")    # 模拟被系统锁定进入归档状态



#例【2-8】
class Prompt:
    def __init__(self, content):
        self.status = "waiting"
        self.content = content

    def activate(self):
        self.status = "active"
        print(f"[状态同步] Prompt已激活：{self.content}")

# 模拟外部事件系统
class ExternalEventSystem:
    def __init__(self):
        self.listeners = []

    def bind(self, callback):
        self.listeners.append(callback)

    def trigger_event(self):
        for callback in self.listeners:
            callback()

# 流程绑定
prompt = Prompt("生成日报内容")
event_system = ExternalEventSystem()

# 外部事件绑定内部Prompt激活逻辑
event_system.bind(prompt.activate)

# 模拟外部事件触发
event_system.trigger_event()



#例【2-9】
class SemanticUnit:
    def __init__(self, text: str):
        self.text = text
        self.role = self._infer_role()
        self.intent = self._map_to_intent()

    def _infer_role(self):
        if "我想" in self.text or "请" in self.text:
            return "user"
        elif "系统" in self.text:
            return "system"
        return "unknown"

    def _map_to_intent(self):
        if "日报" in self.text:
            return "generate_report"
        elif "工具" in self.text:
            return "call_tool"
        return "unknown"

    def to_dict(self):
        return {
            "text": self.text,
            "role": self.role,
            "intent": self.intent
        }

# 示例：自动将语言映射为语义单元
su = SemanticUnit("请帮我生成今天的日报")
print(su.to_dict())



#例【2-10】
# 定义插件注册表
plugin_registry = {}

def register_plugin(intent: str):
    def decorator(func):
        plugin_registry[intent] = func
        return func
    return decorator

# 插件实现：生成日报
@register_plugin("generate_report")
def generate_daily_report(input_text):
    return f"[插件返回] 日报已生成：内容基于输入'{input_text}'分析完成"

# 插件执行器：根据语义节点调用插件
def execute_semantic_plugin(intent: str, input_text: str):
    plugin = plugin_registry.get(intent)
    if plugin:
        return plugin(input_text)
    else:
        return f"[系统提示] 无可用插件处理意图：{intent}"

# 示例调用
intent = "generate_report"
response = execute_semantic_plugin(intent, "请帮我生成今天的日报")
print(response)


